package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo15Cache {
  def main(args: Array[String]): Unit = {
    /**
      * rdd中默认不保存数据，如果对RDD处理多次，这个RDD会处理多次
      *
      * 持久化级别选择
      *
      * １、如果数据量不大，内存充足－－－＞MEMORY_ONLY
      *
      * ２、如果数据量大已经超过内存限制－－－＞MEMORY_AND_DISK_SER（将数据压缩再放入内存）
      *
      * 压缩的好处是体积小可以存放更多的数据，缺点是压缩和解压都需要时间
      */
    //创建配置文件
    val conf: SparkConf = new SparkConf().setAppName("cache").setMaster("local")

    //创建spark对象
    val sc = new SparkContext(conf)

    //读取数据
    val students: RDD[String] = sc.textFile("data/students.txt")

    //取数据
    val studentsRDD: RDD[(String, String, Int, String, String)] = students.map(student => {
      println("studentsRDD处理")
      //切分数据
      val split: Array[String] = student.split(",")
      val id: String = split(0)
      val name: String = split(1)
      val age: Int = split(2).toInt
      val gender: String = split(3)
      val clazz: String = split(4)
      (id,name,age,gender,clazz)
    })
    //students加入缓存
    //studentsRDD.cache()

    //cache默认是MEMORY_ONLY 将数据放入内存

    //MEMORY_AND_DISK_SER 将数据压缩放入内存
    studentsRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)
    //统计班级人数
    studentsRDD.map(stu =>(stu._5,1)).reduceByKey(_+_).foreach(println)

    //统计各个年龄段的人数
    studentsRDD.map(stu =>(stu._3,1)).reduceByKey(_+_).foreach(println)

    //统计性别人数
    studentsRDD.map(stu =>(stu._4,1)).reduceByKey(_+_).foreach(println)
  }
}
